package net.flyinggroup.scs.gitter.repository.impl;

import net.flyinggroup.scs.gitter.domain.Message;
import net.flyinggroup.scs.gitter.domain.User;
import net.flyinggroup.scs.gitter.repository.UserRepository;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.ReactiveMongoOperations;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.stereotype.Service;
import reactor.core.publisher.EmitterProcessor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@Service
public class DefaultUserRepository implements UserRepository {
    private final ReactiveMongoOperations mongoOperations;

    public DefaultUserRepository(ReactiveMongoOperations mongoOperations) {
        this.mongoOperations = mongoOperations;
    }

    @Override
    public Mono<User> findMostActive() {
        return Mono.fromDirect(mongoOperations.aggregate(
                Aggregation.newAggregation(Aggregation.group("user.id")
                                .addToSet("user.name").as("name")
                                .addToSet("user.displayName").as("displayName")
                                .count().as("popularity"),
                        Aggregation.sort(Sort.by(Sort.Direction.DESC, "popularity")),
                        Aggregation.limit(1),
                        Aggregation.unwind("name"),
                        Aggregation.unwind("displayName")
                ), Message.class, User.class));
    }

    @Override
    public Mono<User> findMostPopular() {
        return Mono.fromDirect(mongoOperations.aggregate(
                Aggregation.newAggregation(
                        Aggregation.unwind("mentions"),
                        Aggregation.group("mentions.userId")
                                .addToSet("mentions.screenName").as("name")
                                .addToSet("mentions.screenName").as("displayName")
                                .count().as("popularity"),
                        Aggregation.sort(Sort.by(Sort.Direction.DESC, "popularity")),
                        Aggregation.limit(1),
                        Aggregation.unwind("name"),
                        Aggregation.unwind("displayName")
                ), Message.class, User.class));
    }
}
